package com.scala.spark.typedOperation


import org.apache.spark.sql.SparkSession

object TypedOperation {
    case class Employee(name:String,age:Long,depId:Long,gender:String,salary:Long)
    case class Department(id: Long, name: String)
    def main(args: Array[String]): Unit = {
        val spark = SparkSession.builder().appName("action").master("local").getOrCreate()
        import spark.implicits._
        val employee=spark.read.json("employee")
        val employee2=spark.read.json("employee2")
        val department=spark.read.json("department")
        val departMentDS = department.as[Department]
        val DataSet = employee.as[Employee]
        val DataSet2=employee2.as[Employee]
      /*  println(DataSet.rdd.partitions.size)
        //coalesce和repartition操作  都是重定义分区  区别coalesce只能减少分区数量  而且可以选择不发生shuffle
        // repartiton，可以增加分区，也可以减少分区，必须会发生shuffle，相当于是进行了一次重分区操作

        val repartitionDataSet = DataSet.repartition(10)

        //看下分区
        println(repartitionDataSet.rdd.partitions.size)

        val coalescePartion = repartitionDataSet.coalesce(5)
        println(coalescePartion.rdd.partitions.size)
        DataSet.show()*/


        //去重
        // 都是用来进行去重的，区别在哪儿呢？
        // distinct，是根据每一条数据，进行完整内容的比对和去重
        // dropDuplicates，可以根据指定的字段进行去重


       /* val distinctDataSet = DataSet.distinct()
        distinctDataSet.show()
        val singleDrop=DataSet.dropDuplicates(Seq("name"))
        singleDrop.show()
        val dropDataSet = DataSet.dropDuplicates("name","age")
        dropDataSet.show()*/

        // except：获取在当前dataset中有，但是在另外一个dataset中没有的元素
        // filter：根据我们自己的逻辑，如果返回true，那么就保留该元素，否则就过滤掉该元素
        // intersect：获取两个数据集的交集

       /* val exceptDS = DataSet.except(DataSet2)
        exceptDS.show()
        val filterDS = DataSet.filter(employee=>employee.age>30)
        filterDS.show()
        val intersectDS = DataSet.intersect(DataSet2)
        intersectDS.show()*/

        // map：将数据集中的每条数据都做一个映射，返回一条新数据
        // flatMap：数据集中的每条数据都可以返回多条数据
        // mapPartitions：一次性对一个partition中的数据进行处理
    /*   DataSet.map{my=>(my.name,my.age+1000)}.show()
       departMentDS.flatMap{department=>Seq(Department(department.id+1,department.name+"来自于1"),Department(department.id+2,department.name+"来自于2"))}.show()
        DataSet.mapPartitions{
            allValues =>{
                val returnValue=scala.collection.mutable.ArrayBuffer[(String,Long)]()
                while(allValues.hasNext){
                    val every = allValues.next()
                    returnValue+=((every.name,every.salary+8888))
                }
                returnValue.iterator
            }
        }.show()
*/
/*
        DataSet.join(departMentDS,$"depId"===$"id").foreach(println(_))
        //将两个数据集 连到一起
        DataSet.joinWith(department,$"depId"===$"id").foreach(println(_))
        DataSet.show()*/
        //sort 排序
       // DataSet.sort($"age".desc).show()

        //randomSplit 按照权重 将dataset切割为几个dataset
        val randomSplitDS = DataSet.randomSplit(Array(2,3,5))
        randomSplitDS.foreach(ds=>ds.show())
        //sample 按照指定比例，随机抽取数据
        DataSet.sample(false,0.3).show()


    }

}










